package com.lagou.kafka.demo.controller;

import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.logging.log4j.message.Message;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.concurrent.ExecutionException;

/**
 * @param
 * @author oyz
 * @version 1.0
 * @description:  控制层(同步发送)
 * @date 2022/3/25 21:28
 */
@RestController
public class KafkaSyncProducerController {

    @Autowired
    private KafkaTemplate<Integer,String> template;

    @RequestMapping("/send/sync/{message}")
    public String send(@PathVariable(name ="message") String param){
        ListenableFuture<SendResult<Integer, String>> future = template.send("topic-spring-01", 0, 0, param);
        try {
            // 同步发送
            SendResult<Integer, String> result = future.get();
            RecordMetadata metadata = result.getRecordMetadata();
            System.out.println(String.format("消息的类型为:%s,%s,%s",metadata.partition(),metadata.topic(),metadata.offset()));
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
        return "success";

    }


}
